package com.boot.study.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.boot.study.api.{MySensorSource, SensorReading}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}


object JdbcSinkTest {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 读取文件
    //    val inputSteam: DataStream[String] = env.readTextFile("D:\\WorkSpace\\idea\\Flink\\src\\main\\resources\\sensor.txt")
    //    val dataStream: DataStream[SensorReading] = inputSteam.map(data => {
    //      val arr: Array[String] = data.split(",")
    //      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    //    })
    // 自定义source
    val dataStream: DataStream[SensorReading] = env.addSource(new MySensorSource())
    dataStream.print("jdbc sink")
    dataStream.addSink(new MyJdbcSinkFunction)

    // 执行
    env.execute("jdbc sink test")
  }
}

class MyJdbcSinkFunction extends RichSinkFunction[SensorReading] {
  // 定义连接、预编译语句
  var conn: Connection = _
  var instertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _

  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306/test", "root", "123")
    instertStmt = conn.prepareStatement("insert into sensor_temp (id,temp) values (?, ?)")
    updateStmt = conn.prepareStatement("update sensor_temp set temp = ? where id = ?")
  }

  override def invoke(value: SensorReading, context: SinkFunction.Context[_]): Unit = {
    // 先执行更新操作，查到就更新
    updateStmt.setDouble(1, value.temperature)
    updateStmt.setString(2, value.id)
    updateStmt.execute()

    // 如果更新没有查到数据就执行插入操作
    if (updateStmt.getUpdateCount == 0) {
      instertStmt.setString(1, value.id)
      instertStmt.setDouble(2, value.temperature)
      instertStmt.execute()
    }
  }

  override def close(): Unit = {
    instertStmt.close()
    updateStmt.close()
    conn.close()
  }
}
